package com.crpc.core.watcher;

import com.crpc.common.crpcEnum.CrpcExceptionEnums;
import com.crpc.common.exception.CrpcException;
import com.crpc.core.CrpcBootStrap;
import com.crpc.core.NettyBootstrapInitializer;
import com.crpc.core.loadBalance.LoadBalance;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

import static com.crpc.core.CrpcBootStrap.CHANNEL_MAP;
import static com.crpc.core.CrpcBootStrap.CRPC_REQUEST_THREAD_LOCAL;

/**
 * UpAndDownWatcher is a class that implements the Watcher interface.
 * It listens for events related to the change in child nodes of a specific path.
 * When the child nodes change, it performs the necessary actions such as creating or updating channels,
 * and reloading the load balancer.
 * <p>
 * This class should be used as a listener for ZooKeeper events related to node children changes.
 */
@Slf4j
public class UpAndDownWatcher implements Watcher {
    @Override
    public void process(WatchedEvent event) {
        //如果事件类型为子节点变化
        if (event.getType() == Event.EventType.NodeChildrenChanged) {
            log.info("子节点发生变化======>{}", event.getPath());
            //获取服务名
            String serviceName = getServiceName(event.getPath());
            //从注册中心获取服务地址
            List<InetSocketAddress> inetSocketAddressList = CrpcBootStrap.getInstance().getConfiguration().getRegistryConfig().getRegistry().lookup(serviceName);
            //获取连接池中的缓存
            Map<InetSocketAddress, Channel> channelMap = CrpcBootStrap.CHANNEL_MAP;
            //创建一个临时的连接池缓存
            Map<InetSocketAddress, Channel> channelMapTemp = new ConcurrentHashMap<>(channelMap);
            //清空连接池中的缓存
            channelMap.clear();
            //遍历服务地址，重新建立连接
            for (InetSocketAddress inetSocketAddress : inetSocketAddressList) {
                if (!channelMapTemp.containsKey(inetSocketAddress)) {
                    Channel channel = createChannel(inetSocketAddress);
                    channelMap.put(inetSocketAddress, channel);
                } else {
                    channelMap.put(inetSocketAddress, channelMapTemp.get(inetSocketAddress));
                }

            }
            log.info("当前连接池中的缓存数量========>{}", channelMap.size());
            //获得负载均衡器重新loadBalance
            LoadBalance loadBalance = CrpcBootStrap.getInstance().getConfiguration().getLoadBalance();
            loadBalance.reLoadBalance(serviceName, inetSocketAddressList);
        }
    }

    private Channel createChannel(InetSocketAddress inetSocketAddress) {
        Channel channel;
        //异步操作,获取通道,异步等待提供者发来消息
        CompletableFuture<Channel> channelFuture = new CompletableFuture<>();
        NettyBootstrapInitializer.getBootstrap().connect(inetSocketAddress).addListener(
                (ChannelFutureListener) promise -> {
                    if (promise.isDone()) {
                        //异步获取结果
                        log.info("[动态感知]channel：{}建立连接成功", inetSocketAddress);
                        channelFuture.complete(promise.channel());
                    } else if (!promise.isSuccess()) {
                        log.error("[动态感知]channel：{}建立连接失败", inetSocketAddress);
                        channelFuture.completeExceptionally(promise.cause());
                    }
                });
        try {
            channel = channelFuture.get(3, TimeUnit.SECONDS);
        } catch (Exception e) {
            log.error("获取通道时发生异常");
            e.printStackTrace();
            throw new CrpcException(CrpcExceptionEnums.NETTY_ERROR);
        } finally {
            CRPC_REQUEST_THREAD_LOCAL.remove();
        }
        //如何等待
        CHANNEL_MAP.put(inetSocketAddress, channel);

        if (channel == null) {
            throw new CrpcException(CrpcExceptionEnums.NETTY_ERROR);
        }
        return channel;
    }

    private String getServiceName(String path) {
        String[] split = path.split("/");
        return split[split.length - 1];
    }
}
